package com.youxin.logprocess.controller;

import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.common.PartitionInfo;
import org.apache.log4j.Logger;
/**
 * Kafka监控控制
 * @author yingde.cao
 *
 */

import com.youxin.busfoundation.config.MonitorBaseBean;
import com.youxin.busfoundation.config.MonitorKafkaConfig;
import com.youxin.foundation.lang.BaseException;
import com.youxin.foundation.utils.thread.ScheduleThreadPool;
import com.youxin.logprocess.common.MonitorGlobalConfig;
import com.youxin.logprocess.kafka.KafkaConsumerTask;
public class MonitorKafkaController extends BaseMonitorController{

	private static Logger logger = Logger.getLogger(MonitorKafkaController.class);
	private static MonitorKafkaController controller = new MonitorKafkaController();
	private MonitorKafkaConfig config;
	
	public MonitorKafkaConfig getConfig() {
		return config;
	}

	public void setConfig(MonitorKafkaConfig config) {
		this.config = config;
	}

	public static void initConfig(MonitorKafkaConfig config) {
		controller.setConfig(config);
	}
	
	public static MonitorKafkaController instance(){
		return controller;
	}
	
	@Override
	public MonitorBaseBean getMonitorBean() {
		return config;
	}
	
	/**
	 * 执行方法
	 */
	public void execute() {
		if(config==null)
		{
			throw new BaseException("请配置mointor.kafka的config");
		}
		if (config.getMonitor() != MonitorGlobalConfig.MONITOR_ON) {
			return;
		}
		
		for(String hosts:config.getServers())
		{
			ScheduleThreadPool.scheduleWithFixedDelay(new Runnable() {
				@Override
				public void run() {
					try {
						listKafkaTopics(hosts);
					} catch (Exception e) {
						logger.error("执行任务异常:",e);
						adds(hosts, e);
					}
				}
			}, config.getInterval(), MonitorGlobalConfig.START_DELAY_TIME);
		}		
	}
	
	private void listKafkaTopics(String hosts){
		Properties properties=new Properties();		
    	properties.put("enable.auto.commit", "true");
    	properties.put("auto.commit.interval.ms", "1000");
    	properties.put("session.timeout.ms", "30000");
    	properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    	properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    	String[] urls=hosts.split(",",0);
		for(String url:urls)
		{
			try {
				url=StringUtils.trimToEmpty(url);
				if(url.length()==0){
					continue;
				}
				properties.put("bootstrap.servers", url);
				KafkaConsumerTask kafkaConsumerTask=new KafkaConsumerTask(properties, "log");				
				kafkaConsumerTask.listTopics();
			} catch (Exception e) {
				logger.error("连接kafka失败", e);
				add(url,e);
			}		
		}  	
	}
}
